package com.lishuang.base.ThreadPoolUtil;

import java.util.Collection;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ForkJoinTask;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * @author LiShuang
 * @version 1.0.0
 * @date 2022/11/5 16:11
 */
public class FutureUtils {

    /**
     * 等效于scala的Future.traverse(M<A> list,(A a)->Future<A>)，返回Future<M<A>>
     *
     * @param list
     * @param function
     * @return
     */
    public static <R, T, CT extends Collection<T>, CR extends Collection<R>>
    CompletionStage<CR> traverse(CT list, Function<T, CompletionStage<R>> function) {
        List<CompletionStage<R>> futureResultList =
                list.stream().map(function).collect(Collectors.toList());
        return sequence(futureResultList);
    }

    /**
     * 等效于scala的Future.sequence(M<Future<A>>)，返回Future<M<A>>
     *
     * @param futureList
     * @return
     */
    public static <T, CT extends Collection<CompletionStage<T>>, CR extends Collection<T>>
    CompletionStage<CR> sequence(CT futureList) {
        CompletableFuture[] futures =
                futureList.stream()
                        .map(s -> (CompletableFuture<T>) s)
                        .toArray(CompletableFuture[]::new);
        CompletableFuture<Void> allResult = CompletableFuture.allOf(futures);
        return (CompletionStage<CR>)
                allResult.thenApply(
                        e ->
                                Stream.of(futures)
                                        .map(CompletableFuture::join)
                                        .collect(Collectors.toList()));
    }

    public static <U> CompletableFuture<U> supplyAsync(
            Supplier<U> supplier, PriorityDistributedThreadPoolExecutor executor, int priority) {
        if (supplier == null) throw new NullPointerException();
        CompletableFuture<U> d = new CompletableFuture<U>();
        //executor.execute(new AsyncSupply<U>(d, supplier), priority);
        return d;
    }

    private static final class AsyncSupply<T> extends ForkJoinTask<Void>
            implements Runnable, CompletableFuture.AsynchronousCompletionTask {
        CompletableFuture<T> dep;
        Supplier<T> fn;

        AsyncSupply(CompletableFuture<T> dep, Supplier<T> fn) {
            this.dep = dep;
            this.fn = fn;
        }

        public final Void getRawResult() {
            return null;
        }

        public final void setRawResult(Void v) {}

        public final boolean exec() {
            run();
            return true;
        }

        public void run() {
            CompletableFuture<T> d;
            Supplier<T> f;
            if ((d = dep) != null && (f = fn) != null) {
                dep = null;
                fn = null;
                if (d.getNow(null) == null) {
                    try {
                        d.complete(f.get());
                    } catch (Throwable ex) {
                        d.completeExceptionally(ex);
                    }
                }
            }
        }
    }
}
